After thinking about it more, what we really need in the near future is all about table: create/alter/lookup/drop tables, instead of how the tables are organized, like databases, and how other information is stored, like view/function.

How about we call it TableSupport?

Works for me.

this is something we should decide now. IMO schema and properties are must-have, but others may not. e.g. if a data source uses a path to lookup table, then there is no database/table name to it. And we don't have a story to deal with partitions yet.

For the table metadata, I think we do need partitions. Iceberg creates partitioned tables and I'd like to start getting the DDL operations working. This is why I proposed this metadata in the SPIP a few months ago. We seem to have lazy consensus around it.

You're right about the name. How about I change it to identifier that could be a path or a name?

I updated the last comment because I thought this was referring to CatalogSupport at first. Sorry about the confusion.

I'll just remove name and database. We can add them later when we figure out how we want to handle it. We need partitioning right away, though.

It's great to have an example to show how to use this API, can we add an example to all the methods here?

Are you looking for examples in Javadoc, or an example implementation?

I added an example to the Javadocs:

import TableChange._
val catalog = source.asInstanceOf[TableSupport].catalog()
catalog.alterTable(ident,
    addColumn("x", IntegerType),
    renameColumn("a", "b"),
    deleteColumn("c")
  )

This is great!

Just noticed that these aren't public, but should be because they will be passed to implementations through alterTable.

These should also implement unapply for Scala implementations.

Nevermind, I forgot that these are in an interface so they are automatically public.

And, I'm not sure it's possible to implement unapply in Java. Not even implementing Product works.

why it's expressions? In current Spark we only support PARTITION BY columns.

I recommend reading the proposal SPIP's "Proposed Changes" section, which goes into more detail than this comment can. In short, you're thinking of partitions as columns like Hive tables, but that is a narrow definition that prevents the underlying format from optimizing queries.

Partitions of a table are derived from the column data through some transform. For example, partitioning by day uses a day transform from a timestamp column: day(ts). Hive doesn't keep track of that transform and requires queries to handle it by inserting both ts and day columns. This leads to a few problems, including:

• Hive has no ability to transform ts > X to the partition predicate day >= day(X). Queries that don't take into account the table's physical storage by adding partition predicates by hand will result in full table scans.
• Users can insert any data they choose into the day partition and it is up to them to do it correctly.

Also, consider bucketing. Bucketing is also a transform that is effectively a partitioning of the table's files: bucket=hash(col) % N. The reason why bucketing is handled as a special case in Hive is that using it requires knowing the transform and relationship between the bucket number and its column. If we think of partitioning as grouping data by common values of a set of transforms, then buckets are just another partition that we can use for purposes like bucketed joins or limiting scans when looking for specific values.

If the transform is identity -- just copy the value into partition data -- then you have the same functionality that Hive provides. But by building the transformations into the partitioning layer, we can do more to optimize queries, while hiding the physical layout of a table.

Using Expression allows Spark to pass day(ts) to the data source. It is up to the source which expressions are supported. The current FS tables would reject any expression that isn't just a column reference. Iceberg supports identity, year, month, day, hour, truncate, and bucket transforms.

I read the SPIP but I still can't figure it out. How can Spark pass these "partition transform" to data source? The current end-user API only allows users to specify partition columns.

And why does the "partition transform" belong to a table definition? I think it's reasonable to say that a table is partition by column timestamp, and supports pushing partition predicates like year(timestamp) > 2000.

The current end-user API only allows users to specify partition columns.

I think an example would help understand the use of expression here. Right now, I can create a table partitioned by day like this:

CREATE TABLE t (ts timestamp, data string, day string) PARTITIONED BY (day)

Then it's up to queries to supply the right values for day in their queries. I'm proposing we change that to something like the following that uses expressions in the PARTITIONED BY clause instead of only allowing column names:

CREATE TABLE t (ts timestamp, data string) PARTITIONED BY (date(ts));

This can handle all identity partitioning in Hive tables today and it can handle bucketing.

And why does the "partition transform" belong to a table definition?

Transforms should be passed to the table so the source can use them for the physical layout. In DataSourceV2, the source could be anything so it needs to be the component that handles the physical layout. Because we want distributed data sources, we need some way of telling them how to distribute data.

For example, I could use a partitioning expression to tell a source how to shard across PostgreSQL instances. I could also use it to define the keys in an HBase connector. Those are uses of partitioning that Spark can't handle internally.

Like Hive, Spark has only supported a limited definition of partitioning up to now, but I'd like to be able to put tables using Hive's layout behind this API eventually. I think this way of configuring partitioning is a good way to do that, while supporting what Iceberg and other sources will need.

Another benefit: this would allow us to translate BUCKETED BY clauses into something we can actually pass to data sources.

I see, and this seems a very cool feature. My only concern is that, this new feature is not being discussed in dev list yet, and no JIRA ticket is tracking it. I feel a little weird to support a non-existing feature in data source v2 API. Shall we start a thread in dev list for this new feature? And see if we can make it before 2.4.

I wouldn't say this way of passing partitioning is a new feature. It's just a generalization of the existing partitioning that allows us to pass any type of partition, whether it is bucketing or column-based.

As for open discussion, this was proposed in the SPIP that was fairly widely read and commented on. That SPIP was posted to the dev list a few times, too. I do appreciate you wanting to make sure there's been a chance for the community to discuss it, but there has been plenty of opportunity to comment. At this point, I think it's reasonable to move forward with the implementation.

Note that this is private[sql]. This allows us to use the named TableCatalog instances without solving how multiple catalogs should be exposed to users through a public API just yet.

Can we support adding a comment to a column? / table?

Yeah, we can add that. What do you suggest changing?

is adding or dropping table partitions a table change?

No, this API covers table configuration, not data modification. If you're interested in dropping partitions, you should look at the DeleteSupport API proposal.

Would it be a valid operation to change the partitioning of the table without dropping the entire table and re-creating it? E.g. change the bucket size for such and such column. Seems pretty difficult to do in practice though since the underlying data layout would have to change as part of the modification.

@mccheah, our table format supports updating the partitioning of a table, so I think it should be supported. But, this is intended to be an initial API so I didn't want to block this on agreeing how to repartition a table.

The nomenclature here appears to conflict with @cloud-fan's refactor in https://github.com/apache/spark/pull/22547/files#diff-45399ef5eed5c873d5f12bf0f1671b8fR40. Maybe we can call this TableMetadata or TableDescription? Or perhaps we rename the other construct?

I think the two Table classes are trying to be the same thing. This is one of the reasons why I brought it up in the sync. @cloud-fan's current PR isn't yet based on this work, so it doesn't get the abstraction right.

What you linked to uses Table to expose newScanConfigBuilder, basically requiring that all tables are readable. Instead, the implementation classes in #22547 should be interfaces that extend this Table to make it readable.

Looking at this patch in comparison to the other again (updated to #23086) it looks like this work should be rebased on top of the batch read refactor's PR in order to not have two Table classes that do the same thing - is this the right assessment?

Would it be a valid operation to change the partitioning of the table without dropping the entire table and re-creating it? E.g. change the bucket size for such and such column. Seems pretty difficult to do in practice though since the underlying data layout would have to change as part of the modification.

Does it include a View, like what we are doing in the CatalogTable?

No, this interface carries minimal set of operations needed to implement the v2 logical plans. We can expand it later when we need to.

The goal here is to build a replacement catalog API incrementally and to avoid requiring all catalogs to implement all possible catalog features. This API is focused on table operations, not view or function operations that we have yet to define.

As we discussed, will these APIs now live in the sql-api package? Also at what point are we going to introduce this new Maven module and package?

@cloud-fan, do you want me to create the sql-api package in this PR, or do you want to add a separate PR to move the current v2 API?

The semantics here aren't clear at least to me. Typically a <something>Provider is a class that can instantiate a something. Here it appears to provide itself? The abstraction I would imagine would be to have either:

• CatalogProvider has a method called get(options, sqlConf) which returns a TableCatalog configured with the given options and SQLConf, or
• Remove CatalogProvider entirely and put initialize in this interface. Every TableCatalog instance must be initialized before calling any other methods like loadTable.

The intent is to use some interface to load all catalogs, whether they implement TableCatalog, FunctionCatalog, or both (or other catalog API parts). So you load a catalog, then check whether it is a TableCatalog when you want to use it for tables.

Sounds like the name CatalogProvider is the confusing part. You're right that a provider usually implements a get method to provide something. I could change that to CatalogImpl or something. Would that work?

Perhaps something as simple as Catalog as the top level then, which is sub-interfaced by TableCatalog, FunctionCatalog, and other "kinds" of catalogs. They can all share the same initialize method which is declared by Catalog. That sounds like the simplest idea, perhaps?

What about CatalogPlugin? I'm hesitant to go with just Catalog because it isn't very specific. I think it might cause confusion because the interface has only the initialize method.

I'm ok with that!

Consider .map(...).getOrElse, but we haven't been consistent on using or not using match on Option types throughout Spark in general anyways so it's fine to leave as-is.

Also any particular reason this and the following variables have to be lazy?

How would the getOrElse pattern work here? If the URI is undefined, what tuple should be added to the table properties?

A second read over this, don't think we necessarily have to use the Option lambdas here, and in fact may be less legible, varying from developer to developer.

But if one were to do so, it'd be something like this...

v1Table.storage.properties + v1Table.storage.locationUri.map(uri -> Map("path" -> CatalogUtils.URITOString(uri)).getOrElse(Map.empty)

Is it necessary to support all PartitionTransform types for a first pass? Though, I would imagine we'd have to for the v2 to v1 catalog adapter. If it weren't for that, I would suggest supporting only a simple set of PartitionTransform, such as only identity, to keep this PR focused on the catalog API and not the partitions API.

What I wanted to discuss on Wednesday was how to pass these transforms. @rxin and I had some discussions about it on the dev list, but we didn't come up with a decision. I think the solution will probably be to add way to pass generic function application and a list of arguments that are either columns or constant literals.

I should note that the generic function application will probably look like the Apply case.

Is it possible to defer partition support, or is is fundamentally important enough to get that correct now because we will be building on it on e.g. the very next evolution of this API and its uses? I'm thinking about how to minimize the amount of API we're proposing per change, particularly if choices aren't particularly obvious.

I think we should get this done now. Partition transforms are a generalization of Hive partitioning (which uses some columns directly) and bucketing (which is one specific transform). If we add transformation functions now, we will support both of those with a simple API instead of building in special cases for identity and bucket transforms.

I also have a data source that allows users to configure partitioning using more transforms than just identity and bucketing, so I'd like to get this in so that DDL for those tables works.

Looking at this patch in comparison to the other again (updated to #23086) it looks like this work should be rebased on top of the batch read refactor's PR in order to not have two Table classes that do the same thing - is this the right assessment?

Should we default this to an empty Map? I don't think all tables will support custom properties.

Yeah, that works.

Should we default this to no partitioning?

Sure

I use lazy for a couple reasons. First, to avoid building maps or other data values that are never used. Second, to avoid a required ordering for fields. If fields depend on one another, then they have to be reordered when those dependencies change. Lazy values never require reordering.